package com.sc.passbook.service.impl;

import com.sc.passbook.constant.Constant;
import com.sc.passbook.mapper.PassTemplateMapper;
import com.sc.passbook.service.IGainPassTemplateService;
import com.sc.passbook.utils.RowKeyGen;
import com.sc.passbook.vo.GainPassTemplateRequest;
import com.sc.passbook.vo.PassTemplate;
import com.sc.passbook.vo.Response;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.nio.file.Files;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.Date;

/**
 * @author sc
 * Created on  2018/9/3
 */
@Service
@Slf4j
public class GainPassTemplateServiceImpl implements IGainPassTemplateService {

    @Autowired
    private HbaseTemplate hbaseTemplate;
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public Response gain(GainPassTemplateRequest request) throws Exception {
        String passTemplateId = RowKeyGen.genPassTemplateRowKey(request.getPassTemplate());
        PassTemplate passTemplate = hbaseTemplate.get(Constant.PassTemplateTable.TABLE_NAME,
                passTemplateId, new PassTemplateMapper());
        if(passTemplate.getLimit()<=1&&passTemplate.getLimit()!=-1){
            log.error("error");
        }
        Date curr=new Date();
        if(!(curr.getTime()>=passTemplate.getStart().getTime()&&curr.getTime()<=passTemplate.getEnd().getTime())) {
            log.error("时间不允许");
        }
        if(passTemplate.getLimit()!=-1){
            byte[] FAMILY_C = Bytes.toBytes(Constant.PassTemplateTable.FAMILY_C);
            byte[] LIMIT = Bytes.toBytes(Constant.PassTemplateTable.LIMIT);
            Put p  = new Put(Bytes.toBytes(passTemplateId));
            p.addColumn(FAMILY_C,LIMIT,Bytes.toBytes(passTemplate.getLimit()-1));
            hbaseTemplate.saveOrUpdate(Constant.PassTemplateTable.TABLE_NAME,p);
        }
        if(!addPassForUser(request,passTemplate.getId(),passTemplateId)){
            log.error("fail");
        }
        return Response.success();
    }


    private void recordTokenToFile(Integer merchantsId, String passTemplateId, String token) throws Exception {
        Files.write(
                Paths.get(Constant.TOKEN_DIR, String.valueOf(merchantsId),
                        passTemplateId, Constant.USED_TOKEN_SUFFIX), (token + "\n").getBytes(),
                StandardOpenOption.APPEND
        );

    }

    private boolean addPassForUser(GainPassTemplateRequest request, Integer merchansId, String passTemplateId) throws Exception {
        byte[] FAMILY_I = Bytes.toBytes(Constant.PassTable.FAMILY_I);
        byte[] USER_ID = Bytes.toBytes(Constant.PassTable.USER_ID);
        byte[] TEMPLATE_ID = Bytes.toBytes(Constant.PassTable.TEMPLATE_ID);
        byte[] TOKEN = Bytes.toBytes(Constant.PassTable.TOKEN);
        byte[] ASSIGNED_DATE = Bytes.toBytes(Constant.PassTable.ASSIGNED_DATE);
        byte[] CONSUME_DATE = Bytes.toBytes(Constant.PassTable.CONSUME_DATE);

        Put put = new Put(Bytes.toBytes(RowKeyGen.genPassRowKey(request)));
        put.addColumn(FAMILY_I, USER_ID, Bytes.toBytes(request.getUserId()));
        put.addColumn(FAMILY_I, TEMPLATE_ID, Bytes.toBytes(passTemplateId));
        if (request.getPassTemplate().getHasToken()) {
            String token = redisTemplate.opsForSet().pop(passTemplateId);
            if (StringUtils.isBlank(token)) {
                log.error("token not found " + passTemplateId);
                return false;
            }
            recordTokenToFile(merchansId, passTemplateId, token);
            put.addColumn(FAMILY_I, TOKEN, Bytes.toBytes(token));
        } else {
            put.addColumn(FAMILY_I, TOKEN, Bytes.toBytes("-1"));
        }
        put.addColumn(FAMILY_I, ASSIGNED_DATE, Bytes.toBytes(DateFormatUtils.format(new Date(), "yyyy-MM-dd")));
        put.addColumn(FAMILY_I, CONSUME_DATE, Bytes.toBytes("-1"));
        hbaseTemplate.saveOrUpdate(Constant.PassTable.TABLE_NAME, put);
        return true;
    }
}
